package com.msl.debezium.kafka.producer;

import com.alibaba.fastjson.JSON;
import com.msl.debezium.common.BizException;
import com.msl.debezium.common.Msg;
import com.msl.debezium.common.ResultCodeEnum;
import com.msl.debezium.dto.UserDto;
import io.swagger.annotations.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/producer")
@Api(value = "发送kafka消息", tags = {"KafkaSender"})
@ApiResponses({@ApiResponse(code = 200, message = "OK,请求成功"),@ApiResponse(code = 400, message = "bad-request,客户端请求错误")
              ,@ApiResponse(code = 500, message = "服务器内部错误"),@ApiResponse(code = 403, message = "未授权访问")})
public class KafkaSender {

    private Logger logger = LoggerFactory.getLogger(KafkaSender.class);

    private static final String TOPICS = "mysql.kang.user";

    @Autowired
    private KafkaTemplate template;

    @PostMapping("/user/sendMq")
    @ApiOperation(value = "sendMq",notes = "用户信息新增的数据推送到消息队列")
    public Msg<String> sendMq(@RequestBody UserDto user){
        Msg<String> msg = Msg.ok();
        try{
            String message = JSON.toJSONString(user);
            logger.info("向kafka发送消息："+message);
            template.send(TOPICS, message);
        }catch(Exception e){
            Msg.err(ResultCodeEnum.FAIL, "发送kafka消息失败");
        }
        return msg;
    }

}
